/*
 * To change this template, choose Tools | Templates
 * and open the template in the editor.
 */
package wmr.reverts;


import java.io.IOException;
import java.util.*;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.*;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;
import wmr.core.*;
import wmr.util.Utils;

/**
 *
 * @author Shilad Sen
 * 
 */
public class RevertGraphGenerator extends Configured implements Tool {
    static final int WINDOW_SIZE = 5;

    /*
     * Takes key-value 7zip hashes and outputs url-id pairs.
     */
    public static class MyMapper extends Mapper<Long, AllRevisions, Text, Text> {

        @Override
        public void map(Long pageId, AllRevisions revs, Mapper.Context context) throws IOException, InterruptedException {
            Page article = revs.getPage();
            context.progress();
            if (article.isNormalPage()) {
                LinkedList<Long> windowPrints = new LinkedList<Long>();
                LinkedList<Revision> windowRevs = new LinkedList<Revision>();
                for (Revision rev : revs.getRevisions()) {
                    context.progress();
                    if (rev == null) {
                        continue;
                    }

                    Long fingerprint = rev.getTextFingerprint();
                    String code = null;
                    if (rev.isVandalismStrictRevert()) {
                        code = "vs";
                    } else if (rev.isVandalismLooseRevert()) {
                        code = "vl";
                    } else if (windowPrints.contains(fingerprint)) {
                        code = "f";
                    }
                    if (code != null && !windowRevs.isEmpty()) {
                        int i = windowPrints.indexOf(fingerprint);
                        Revision reverted = windowRevs.get((i == -1) ? 0 : i);
                        if (rev.getContributor().isBot() || reverted.getContributor().isBot()) {
                            code = "b" + code;
                        }
                        if (rev.getText().length() < 60) {
                            code = "s" + code;
                        }
                        context.write(
                                new Text(pageId + "@" + article.getName()),
                                new Text(code + "\t" + formatRevision(rev) + "\t" + formatRevision(reverted))
                            );
                    }
                    windowPrints.add(0, fingerprint);
                    windowRevs.add(0, rev);
                    if (windowPrints.size() > WINDOW_SIZE) {
                        windowPrints.remove(windowPrints.size() - 1);
                        windowRevs.remove(windowRevs.size() - 1);
                    }
                }
            }
        }
    }

    public static String formatRevision(Revision rev) {
        String comment = rev.getComment();
        if (comment == null || comment.length() == 0) comment = "null";
        comment = Utils.cleanupString(comment, 250);
        return rev.getId() + "\t"
             + rev.getTimestamp() + "\t"
             + rev.getContributor().getId() + "@" + rev.getContributor().getName() + "\t"
             + comment;
    }

    public int run(String args[]) throws Exception {

        if (args.length < 2) {
            System.out.println("usage: [input output]");
            ToolRunner.printGenericCommandUsage(System.out);
            return -1;
        }

        Path inputPath = new Path(args[0]);
        Path outputPath = new Path(args[1]);

        Configuration conf = getConf();
        Job job = new Job(conf, this.getClass().toString());

        FileInputFormat.setInputPaths(job, inputPath);
        FileOutputFormat.setOutputPath(job, outputPath);

        
        job.setJarByClass(RevertGraphGenerator.class);
        job.setInputFormatClass(AllRevisionsInputFormat.class);
        job.setOutputFormatClass(TextOutputFormat.class);
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(Text.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(Text.class);
        
        job.setMapperClass(MyMapper.class);
        job.setReducerClass(Reducer.class); // identity reducer
        FileSystem hdfs = FileSystem.get(outputPath.toUri(), conf);
        if (hdfs.exists(outputPath)) {
            hdfs.delete(outputPath, true);
        }

        return job.waitForCompletion(true) ? 0 : 1;
    }

    /**
     * Dispatches command-line arguments to the tool via the
     * <code>ToolRunner</code>.
     */
    public static void main(String[] args) throws Exception {
        int res = ToolRunner.run(new RevertGraphGenerator(), args);
        System.exit(res);
    }
}
